package org.dubnation.user.rabbitmq.handler;

import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.dubnation.base.enums.ResponseCode;
import org.dubnation.base.exception.BusinessException;
import org.dubnation.base.util.SendEmailUtil;
import org.dubnation.base.util.verifycode.SendPhoneSmsUtil;
import org.dubnation.user.controller.AmqpTestController;
import org.dubnation.user.domain.User;
import org.dubnation.user.rabbitmq.config.RegisterNoticeRabbitmqConfig;
import org.dubnation.user.rabbitmq.entity.SendEntity;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;

/**
 * 监听并消费消息
 */
@Component
@Slf4j
public class NoticeHandler {

    @Autowired
    private RedisTemplate<Object, Object> redisTemplate;

    /**
     * 接收邮件消息
     * @param sendEntity
     */
    @RabbitListener(queues = RegisterNoticeRabbitmqConfig.QUEUE_INFORM_EMAIL)
    public void receiveEmail(SendEntity sendEntity) {
        SendEmailUtil.sendEmail(sendEntity);
    }

    /**
     * 接收短信消息
     * @param sendEntity
     */
    @RabbitHandler
    @RabbitListener(queues = RegisterNoticeRabbitmqConfig.QUEUE_INFORM_SMS)
    public void receiveSms(SendEntity sendEntity) {
        try {
            SendPhoneSmsUtil.sendSms(sendEntity.getTo(), sendEntity.getContent());
        } catch (Exception e) {
            e.printStackTrace();
            throw new BusinessException(ResponseCode.RESPONSE_CODE_4000028);
        }
    }

    @RabbitHandler
    @RabbitListener(queues = RegisterNoticeRabbitmqConfig.QUEUE_TEST)
    public void receiveTestMsg(Message message, Channel channel) {
        //监听队列消费消息
        // 消息序号
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        System.out.println(deliveryTag);
        //取出消息唯一标识
        String messageId = message.getMessageProperties().getHeader("spring_returned_message_correlation");
        System.out.println(messageId);
        // 取出消息
        String userStr = new String(message.getBody(), StandardCharsets.UTF_8);
        User user = JSONObject.parseObject(userStr, User.class);
        try {
            //幂等性校验-根据消息id查询该消息是否被消费
            String uuid = UUID.randomUUID().toString().replace("-", "");
            Boolean msgConsumeOrNot = redisTemplate.opsForValue().setIfAbsent(messageId, uuid);
            if (Boolean.TRUE.equals(msgConsumeOrNot)) {
                //设置成功, 该条消息还没有被消费, 正常消费该消息
                System.out.println(user);
                log.info("====消息处理完毕, 用户信息:{}====", user);
            } else {
                //消息已经被消费
                log.info("====消息已经被消费===");
            }
            channel.basicAck(deliveryTag, false); //false表示只确认当前消息
        } catch (Exception e) {
            //发生异常
            try {
                //multiple=true-拒绝当前消息, multiple=false-拒绝当前消息之前的所有消息
                //requeue=true-退回消息, requeue=false-开启死信队列, 该条消息不能再被消费
                channel.basicNack(deliveryTag, true, false);
                log.error("====消息消费失败, 用户信息:{}, 原因是:{}====", user, e.getMessage());
            } catch (IOException ex) {
                throw new RuntimeException(ex);
            }
        }
        log.info("====消费者处理完毕====");
    }
}
